package com.springcloud.biz;

import com.springcloud.topic.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 顾志杰
 * @date 2020/7/29-10:26
 */
@Slf4j
@EnableBinding(value = {Sink.class,
        MyTopic.class,
        GroupTopic.class,
        DelayedTopic.class,
        ErrorTopic.class,
        RequeueTopic.class,
        DlqTopic.class,
        FallbackTopic.class
})
public class StreamConsumer {

    private final AtomicInteger count = new AtomicInteger(1);

    /**
     * 系统自带消费
     *
     * @param payload message
     */
    @StreamListener(Sink.INPUT)
    public void consume(Object payload) {
        log.info("message consume successfully, payload = {}", payload);
    }


    /**
     * 自定义消费
     *
     * @param payload message
     */
    @StreamListener(MyTopic.INPUT)
    public void consumeMyMessage(Object payload) {
        log.info("My message consume successfully, payload = {}", payload);
    }

    /**
     * 测试分组消费
     *
     * @param payload message
     */
    @StreamListener(GroupTopic.INPUT)
    public void consumeGroupMessage(Object payload) {
        log.info("Group message consume successfully, payload = {}", payload);
    }

    /**
     * 测试延时消费
     *
     * @param payload message
     */
    @StreamListener(DelayedTopic.INPUT)
    public void consumeDelayedMessage(MessageBean payload) {
        log.info("Delayed message consume successfully, payload = {}", payload.getPayload());
    }

    /**
     * 异常重试（单机）
     *
     * @param payload message
     */
    @StreamListener(ErrorTopic.INPUT)
    public void consumeErrorMessage(MessageBean payload) {
        log.info("Are you OK?");
        if (count.incrementAndGet() % 3 == 0) {
            log.info("Fine, thank you");
            count.set(0);
        } else {
            log.info("What's your problem");
            throw new RuntimeException("I’m not OK");
        }
        log.info("Delayed message consume successfully, payload = {}", payload.getPayload());
    }

    /**
     * 异常重试（联机版-重新入列）
     *
     * @param payload message
     */
    @StreamListener(RequeueTopic.INPUT)
    public void requeueErrorMessage(MessageBean payload) {
        log.info("Are you OK?");
        try {
            Thread.sleep(3000L);
        } catch (Exception e) {
            e.printStackTrace();
        }
        throw new RuntimeException("I’m not OK");
    }


    /**
     * 死信队列
     *
     * @param payload message
     */
    @StreamListener(DlqTopic.INPUT)
    public void consumeDlqMessage(MessageBean payload) {
        log.info("Dlq - Are you OK?");
        if (count.incrementAndGet() % 3 == 0) {
            log.info("Dlq - Fine, thank you");
        } else {
            log.info("Dlq - What's your problem");
            throw new RuntimeException("I’m not OK");
        }
        log.info("Dlq - Delayed message consume successfully, payload = {}", payload.getPayload());
    }

    /**
     * fallback + 升版
     *
     * @param payload message
     */
    @StreamListener(FallbackTopic.INPUT)
    public void goodbyeBadGuy(MessageBean payload,
                              @Header("version") String version) {
        log.info("fallback - Are you OK?");
        if ("1.0".equalsIgnoreCase(version)) {
            log.info("fallback - Fine, thank you");
        } else if ("2.0".equalsIgnoreCase(version)) {
            log.info("unsupported version");
            throw new RuntimeException("I'm not OK");
        } else {
            log.info("fallback - version = {}", version);
        }
    }

    @ServiceActivator(inputChannel = "fallback-topic.fallback-group.errors")
    public void fallback(Message<?> message){

       log.info("fallback entered");
    }


}
